package org.thingsboard.server.dao.timeseries;

import org.apache.commons.lang.StringUtils;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;
import org.influxdb.dto.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.thingsboard.server.dao.util.CamelConversion;

import javax.annotation.PostConstruct;
import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author: sanhoitou[sanhoitou@hotmail.com]
 * @Date: 2021/5/25
 */
public class AbstractInfluxBaseTimeseriesDao {

    private static final Logger logger = LoggerFactory.getLogger(AbstractInfluxBaseTimeseriesDao.class);

    @Value("${influx.host}")
    private String host;

    @Value("${influx.port}")
    private String port;

    /**
     * 用户名
     */
    @Value("${influx.user}")
    private String username;
    /**
     * 密码
     */
    @Value("${influx.password}")
    private String password;
    /**
     * 数据库
     */
    @Value("${influx.database}")
    private String database;
//    /**
//     * 保留策略
//     */
//    @Value("${influx.retentionPolicy}")
//    private String retentionPolicy;

    private InfluxDB influxDB;

    @PostConstruct
    public void init() {
//        List<String> serverAddressList = new ArrayList<>();
//        for (String host : hosts.split(",")) {
//            serverAddressList.add(String.format("%s:%s", host, port));
//        }
        String serverAddress = String.format("http://%s:%s", host, port);
        influxDB = InfluxDBFactory.connect(serverAddress, username, password);

        try {
            // 如果指定的数据库不存在，则新建一个新的数据库，并新建一个默认的数据保留规则
            if (!this.databaseExist(database)) {
                createDatabase(database);
//                createRetentionPolicy();
                //alter retention policy autogen on thingsboard duration 366d  REPLICATION 1  SHARD DURATION 1d default
            }
            influxDB.setDatabase(database);
        } catch (Exception e) {
            // 该数据库可能设置动态代理，不支持创建数据库
            logger.error("初始化tsdb时，[TsdbService]发生错误，错误消息: {}", e);
        } finally {
//            influxDB.setRetentionPolicy(retentionPolicy == null || "".equals(retentionPolicy) ? "autogen" : retentionPolicy);
            if(this.ping()){
                logger.info("·····连接influx时序数据库成功");
            }
        }

        influxDB.setLogLevel(InfluxDB.LogLevel.NONE);

        //每隔1000点刷新一次，至少每100ms刷新一次
        //bufferLimit表示可以在重试缓冲区中存储的最大点数
        //exceptionHandler表示处理异步错误的消费者函数
        //threadFactory表示要使用的ThreadFactory实例
        influxDB.enableBatch(BatchOptions.DEFAULTS
                .actions(1000)
                .flushDuration(100)
                .bufferLimit(10)
                .exceptionHandler((points, e) -> {
                    List<Point> target = new ArrayList<>();
                    points.forEach(target::add);
                    String msg = String.format("failed to write points:%s\n", target.toString().substring(0, 10000));
                    logger.error(msg, e);
                })
                .threadFactory(
                        Executors.defaultThreadFactory()
                ));
    }

    /**
     * 测试连接是否正常
     *
     * @return true 正常
     */
    public boolean ping() {
        boolean isConnected = false;
        Pong pong;
        try {
            pong = influxDB.ping();
            if (pong != null) {
                isConnected = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isConnected;
    }

    public void createDatabase(String database) {
        influxDB.query(new Query("CREATE DATABASE " + database, ""));
    }

    public void dropDatabase(String database) {
        influxDB.query(new Query("DROP DATABASE " + database, ""));
    }

    public boolean databaseExist(String database) {
        return influxDB.databaseExists(database);
    }

    public void createRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                "default_policy", database, "1h", 3);
        this.query(command);
    }

    public void createRetentionPolicy(String database, String policyName, String duration, int replication, Boolean isDefault) {
        String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
                database, duration, replication);
        if (isDefault) {
            sql = sql.concat(" DEFAULT");
        }
        this.query(sql);
    }

//    public void dropRetentionPolicy() {
//        this.dropRetentionPolicy(database, retentionPolicy);
//    }

    public void dropRetentionPolicy(String database, String retentionPolicy) {
        String sql = String.format("DROP RETENTION POLICY %s ON %s", retentionPolicy, database);
        this.query(sql);
    }

//    @Override
//    public void createContinuousQuery(String measurement) {
//        String cqName = String.format("cq_%s", measurement);
//        String originMeasurement = String.format("%s.%s.%s", database, retentionPolicy, measurement);
//        String cqMeasurement = String.format("%s.%s.%s_hour", database, extendPolicy, measurement);
//        String sql = String.format("CREATE CONTINUOUS QUERY \"%s\" ON %s RESAMPLE EVERY 1h FOR 2h BEGIN SELECT MEAN(*) INTO %s FROM %s GROUP BY time(1h),* FILL(none) END",
//                cqName, database, cqMeasurement, originMeasurement);
//
//        this.query(sql);
//    }

    public boolean continuousQueryExists(String measurement) {
        String cqName = String.format("cq_%s", measurement);
        return continuousQueryExists(database, cqName);
    }

    public boolean continuousQueryExists(String database, String cqName) {
        String sql = "SHOW CONTINUOUS QUERIES";
        QueryResult result = query(sql);
        List<QueryResult.Series> seriesList = result.getResults().get(0).getSeries();
        if (seriesList != null) {
            for (QueryResult.Series series : seriesList) {
                if (database.equals(series.getName())) {
                    List<List<Object>> continuousQueryList = series.getValues();
                    if (continuousQueryList == null) {
                        return false;
                    } else {
                        for (List<Object> queryResult : continuousQueryList) {
                            if (cqName.equals(queryResult.get(0))) {
                                return true;
                            }
                        }
                    }

                }
            }
        }
        return false;
    }

    public void dropContinuousQuery(String databaseName, String cqName) {
        String sql = String.format("DROP CONTINUOUS QUERY %s ON %s", cqName, databaseName);
        QueryResult result = query(sql);
    }

    public boolean measurementsExists(String measurement) {
        return measurementsExists(database, measurement);
    }

    public boolean measurementsExists(String database, String measurement) {
        String sql = String.format("SHOW MEASUREMENTS ON %s", database);
        QueryResult result = query(sql);
        if (result != null) {
            List<QueryResult.Series> seriesList = result.getResults().get(0).getSeries();
            if (seriesList != null) {
                QueryResult.Series series = seriesList.get(0);
                List<List<Object>> valueList = series.getValues();
                for (List<Object> value : valueList) {
                    if (measurement.equals(value.get(0))) {
                        return true;
                    }
                }
            }
        }
        return false;
    }

    public QueryResult query(String command) {
        return influxDB.query(new Query(command, database));
    }

    public QueryResult dataQuery(String command) {
        return influxDB.query(new Query(command, database), TimeUnit.MILLISECONDS);
    }

    /**
     * 利用java的反射机制进行新增操作
     * @param obj
     * @return
     */
    public Point.Builder reflectToInsert(Object obj){
        //获取度量
        Class<?> clasz = obj.getClass();
        Measurement measurement = clasz.getAnnotation(Measurement.class);
        //构建
        Point.Builder builder = Point.measurement(measurement.name());
        // 获取对象属性
        Field[] fieldArray = clasz.getDeclaredFields();
        Column column = null;
        for(Field field : fieldArray){
            try {
                column = field.getAnnotation(Column.class);
                //设置属性可操作
                field.setAccessible(true);
                if(column.tag()){
                    //tag属性只能存储String类型
                    builder.tag(column.name(), field.get(obj)==null?"":field.get(obj).toString());
                }else{
                    //设置field
                    if(field.get(obj) != null){
                        builder.addField(column.name(), field.get(obj)==null?"":field.get(obj).toString());
                    }
                }
            } catch (IllegalArgumentException | IllegalAccessException e) {
                e.printStackTrace();
            }
        }
        return builder;
    }
    /**
     * 新增单条记录
     */
    public void insertOne(Object obj){
        Point.Builder builder = reflectToInsert(obj);
        influxDB.write(builder.build());
    }

    /**
     * 批量新增,方法一
     */
    public void insertBatchByRecords(List<?> records){
        List<String> lines = new ArrayList<String>();
        records.forEach(record->{
            Point.Builder builder = reflectToInsert(record);
            lines.add(builder.build().lineProtocol());
        });
        influxDB.write(lines);
    }

    /**
     * 批量新增，方法二
     */
    public void insertBatchByPoints(List<?> records){
        BatchPoints batchPoints = BatchPoints.database(database)
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
        records.forEach(record->{
            Point.Builder builder = reflectToInsert(record);
            batchPoints.point(builder.build());
        });
        influxDB.write(batchPoints);
    }

    /**
     * 查询，返回Map集合
     * @param query 完整的查询语句
     * @return
     */
    public List<Object> fetchRecords(String query){
        List<Object> results = new ArrayList<Object>();
        QueryResult queryResult = influxDB.query(new Query(query, database));
        queryResult.getResults().forEach(result->{
            result.getSeries().forEach(serial->{
                List<String> columns = serial.getColumns();
                int fieldSize = columns.size();
                serial.getValues().forEach(value->{
                    Map<String,Object> obj = new HashMap<String,Object>();
                    for(int i=0;i<fieldSize;i++){
                        obj.put(columns.get(i), value.get(i));
                    }
                    results.add(obj);
                });
            });
        });
        return results;
    }

    /**
     * 查询，返回map集合
     * @param fieldKeys 查询的字段，不可为空；不可为单独的tag
     * @param measurement 度量，不可为空；
    //     * @param order
    //     * @param limit
     * @return
     */
    public List<Object> fetchRecords(String fieldKeys, String measurement){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
        query.append(" tz('Asia/Shanghai')");
        return this.fetchRecords(query.toString());
    }

    /**
     * 查询，返回map集合
     * @param fieldKeys 查询的字段，不可为空；不可为单独的tag
     * @param measurement 度量，不可为空；
     * @param order
    //     * @param limit
     * @return
     */
    public List<Object> fetchRecords(String fieldKeys, String measurement, String order){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
        query.append(" order by ").append(order);
        query.append(" tz('Asia/Shanghai')");
        return this.fetchRecords(query.toString());
    }

    /**
     * 查询，返回map集合
     * @param fieldKeys 查询的字段，不可为空；不可为单独的tag
     * @param measurement 度量，不可为空；
     * @param order
     * @param limit
     * @return
     */
    public List<Object> fetchRecords(String fieldKeys, String measurement, String order, String limit){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
        query.append(" order by ").append(order);
        query.append(limit);
        query.append(" tz('Asia/Shanghai')");
        return this.fetchRecords(query.toString());
    }

    /**
     * 查询，返回对象的list集合
     //     * @param query
     * @return
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public <T> List<T> fetchResults(String query, Class<?> clasz){
        List results = new ArrayList<>();
        QueryResult queryResult = influxDB.query(new Query(query, database));
        queryResult.getResults().forEach(result->{
            if(result.getSeries()==null){
                return;
            }
            result.getSeries().forEach(serial->{
                List<String> columns = serial.getColumns();
                int fieldSize = columns.size();
                serial.getValues().forEach(value->{
                    Object obj = null;
                    try {
                        obj = clasz.newInstance();
                        for(int i=0;i<fieldSize;i++){
                            if(null==value.get(i)){
                                continue;
                            }
                            String fieldName = columns.get(i);
                            Field field = clasz.getDeclaredField(CamelConversion.underlineToCamel(fieldName));
                            field.setAccessible(true);
                            Class<?> type = field.getType();
                            if(type == Boolean.class){
                                field.set(obj, Boolean.valueOf(value.get(i).toString()));
                            }else if(type == Double.class){
                                field.set(obj, Double.valueOf(value.get(i).toString()));
                            }
                            else if(fieldName.equals("time")){
                                //时间格式化为yyyy-MM-dd HH:mm:ss
                                String str = value.get(i).toString().replaceAll("T"," ");
                                field.set(obj, StringUtils.substringBefore(str, "."));
                            }
                            else if(type == Long.class){
                                field.set(obj, Long.valueOf(value.get(i).toString()));
                            }else{
                                field.set(obj, value.get(i));
                            }
                        }
                    } catch (NoSuchFieldException | SecurityException | InstantiationException | IllegalAccessException e) {
                        e.printStackTrace();
                    }
                    results.add(obj);
                });
            });
        });
        return results;
    }

//    /**
//     * 查询，返回对象的list集合
//     * @param fieldKeys
//     * @param measurement
//     * @param clasz
//     * @return
//     */
//    public <T> List<T> fetchResults(String fieldKeys, String measurement, Class<?> clasz , String wheres){
//        StringBuilder query = new StringBuilder();
//        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
//        query.append(" where 1=1");
//        if(StringUtils.isBlank(wheres)){
//            List<String> whereList = Arrays.asList(wheres);
//            for (String where : whereList) {
//                query.append(" AND "+where);
//            }
//            query.append(wheres);
//        }
//        query.append(" tz('Asia/Shanghai')");
//        return this.fetchResults(query.toString(), clasz);
//    }
//
//    /**
//     * 查询，返回对象的list集合
//     * @param fieldKeys
//     * @param measurement
//     * @param order
//     * @param clasz
//     * @return
//     */
//    public <T> List<T> fetchResults(String fieldKeys, String measurement, String order, Class<?> clasz, String wheres){
//        StringBuilder query = new StringBuilder();
//        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
//        query.append(" where 1=1");
//        if(StringUtils.isBlank(wheres)){
//            List<String> whereList = Arrays.asList(wheres);
//            for (String where : whereList) {
//                query.append(" AND "+where);
//            }
//            query.append(wheres);
//        }
//        query.append(" order by ").append(order);
//        query.append(" tz('Asia/Shanghai')");
//        return this.fetchResults(query.toString(), clasz);
//    }

    /**
     * 查询，返回对象的list集合
     * @param fieldKeys
     * @param measurement
     * @param order
     * @param limit
     * @param clasz
     * @return
     */
    public <T> List<T> fetchResults(String fieldKeys, String measurement, String wheres,String startTime,String endTime, String order, String limit, Class<?> clasz){
        StringBuilder query = new StringBuilder();
        query.append("SELECT ").append(fieldKeys).append(" FROM ").append(measurement);
        query.append(" WHERE 1=1");
        if (!StringUtils.isBlank(startTime) && !StringUtils.isBlank(endTime)) {
            String formatTime = String.format(" AND \"time\">=\'%s\' AND \"time\"<=\'%s\'", startTime, endTime);
            query.append(formatTime);
        }
        if(!StringUtils.isBlank(wheres)){
            List<String> whereList = Arrays.asList(wheres.split(","));
            for (String where : whereList) {
                query.append(" AND "+ where);
            }
        }
        if(!StringUtils.isBlank(order)){
            query.append(" ORDER BY ").append(order);
        }
        if(!StringUtils.isBlank(limit)){
            query.append(" LIMIT ").append(limit);
        }
        query.append(" tz('Asia/Shanghai')");
        return this.fetchResults(query.toString(), clasz);
    }

    /**
     * 按条件删除
     * @return
     */
    public void deleteByCondition( String measurement, String wheres , String startTime , String endTime){
        if(!StringUtils.isBlank(wheres)){
            StringBuilder query = new StringBuilder();
            query.append("delete ").append(" from ").append(measurement);
            query.append(" WHERE 1=1");
            if (!StringUtils.isBlank(startTime) && !StringUtils.isBlank(endTime)) {
                String formatTime = String.format(" AND \"time\">=\'%s\' AND \"time\"<=\'%s\'", startTime, endTime);
                query.append(formatTime);
            }
            if(!StringUtils.isBlank(wheres)){
                List<String> whereList = Arrays.asList(wheres.split(","));
                for (String where : whereList) {
                    query.append(" AND "+ where);
                }
            }
            influxDB.query(new Query(query.toString(), database));
        }
    }

//    public void batchQueryTest() {
//        long beginTime = 1559613845000L;
//        long endTime = 1559621045000L;
//
//        Map<String, String> dimensionMap = new HashMap<>();
//        dimensionMap.put("tag1", "tag1");
//
//        // 带时间范围的查询
//        String condition = TSDBUtil.getQueryCondition(dimensionMap,beginTime,endTime);
//
//        Object[] args = new Object[]{MEASUREMENT_NAME, condition, QUERY_LIMIT};
//        String command = String.format("SELECT * FROM %s WHERE %s ORDER BY time ASC LIMIT %d", args);
//
//        // 执行查询
//        QueryResult results = tsdbService.dataQuery(command);
//        if (results.getResults() == null) {
//            System.out.println("Data is empty");
//            return;
//        }
//        //results.getResults()是同时查询多条SQL语句的返回值
//        for (QueryResult.Result result : results.getResults()) {
//            List<QueryResult.Series> series = result.getSeries();
//            for (QueryResult.Series serie : series) {
//                List<List<Object>> values = serie.getValues();//字段字集合
//                List<String> colums = serie.getColumns();//字段名
//
//                // 打印查询结果
//                System.out.println("colums:" + colums);
//                for (List<Object> value : values) {
//                    System.out.println("value:" + value);
//                }
//
//                // 封装查询结果
//                List<Map<String, Object>> dataList = new LinkedList<>();
//                for (int i=0;i<values.size();++i){
//                    Map<String, Object> dataMap=new HashMap<>(colums.size());
//                    for (int j=0;j<colums.size();++j){
//                        dataMap.put(colums.get(j),values.get(i).get(j));
//                    }
//                    dataList.add(dataMap);
//                }
//
//                // dataList即可作为返回给用户的查询数据的基础格式
//                System.out.println(dataList);
//
//            }
//        }
//    }


}
